package com.gitee.ixtf.rsocket.kotlinx

import com.gitee.ixtf.rsocket.IRSocketAgent
import io.cloudevents.CloudEvent
import io.rsocket.Payload
import io.vertx.core.Vertx
import kotlinx.coroutines.reactive.asFlow
import kotlinx.coroutines.reactor.awaitSingle
import kotlinx.coroutines.reactor.awaitSingleOrNull
import kotlinx.coroutines.reactor.mono
import org.reactivestreams.Publisher
import reactor.core.publisher.Mono

private lateinit var AGENT: IRSocketAgent

internal fun Vertx.rSocketAgent(agent: IRSocketAgent) {
  AGENT = agent
}

suspend fun Vertx.fireAndForget(event: CloudEvent, route: String) {
  AGENT.rSocketClient().fireAndForget(mono { event.payload(route) }).awaitSingleOrNull()
}

suspend fun Vertx.metadataPush(payloadMono: Mono<Payload>) {
  AGENT.rSocketClient().metadataPush(payloadMono).awaitSingleOrNull()
}

suspend fun Vertx.fireAndForget(payloadMono: Mono<Payload>) {
  AGENT.rSocketClient().fireAndForget(payloadMono).awaitSingleOrNull()
}

suspend fun Vertx.requestResponse(payloadMono: Mono<Payload>) =
    AGENT.rSocketClient().requestResponse(payloadMono).awaitSingle()

suspend fun Vertx.requestStream(payloadMono: Mono<Payload>) =
    AGENT.rSocketClient().requestStream(payloadMono).asFlow()

suspend fun Vertx.requestChannel(payloadMono: Publisher<Payload>) =
    AGENT.rSocketClient().requestChannel(payloadMono).asFlow()
